/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.engine.executor;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.semidata.rtd.core.service.Service;
import com.semidata.rtd.engine.job.Task;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class TaskExecutor implements Service {

	private static final int MAX_TASK_QUEUE_SIZE = 1000000;

	private TaskWorker[] workers;
	private ReentrantLock lock;
	private Condition cond;
	private ReentrantLock taskLock;
	private boolean quit = false;
	private Thread thread = null;
	private Queue<TaskWrapper> tasks = new LinkedList<>();
	private int maxQueueSize;
	private TaskWorkerManager manager;

	public TaskExecutor(int threadNum) {
		this(threadNum, MAX_TASK_QUEUE_SIZE);
	}

	public TaskExecutor(int threadNum, int maxQueueSize) {
		this.workers = new TaskWorker[threadNum];
		this.maxQueueSize = maxQueueSize;
		this.lock = new ReentrantLock();
		this.cond = lock.newCondition();
		this.taskLock = new ReentrantLock();

	}

	public void submit(Task task) {
		boolean flag = false;

		taskLock.lock();
		try {
			if (tasks.size() > maxQueueSize) {
				flag = true;
			} else {
				tasks.add(new TaskWrapper(task));
			}
		} finally {
			taskLock.unlock();
		}

		if (flag) {
			task.afterExecution(new TaskExecutorException(
			        "task queue exceeds max size"));
		} else {
			signal();
		}
	}

	public void signal() {
		lock.lock();
		try {
			cond.signal();
		} finally {
			lock.unlock();
		}
	}

	@Override
	public void start() {

		for (int i = 0; i < workers.length; i++) {
			workers[i] = new TaskWorker("TaskWorker" + i);
			workers[i].start();
		}

		manager = new TaskWorkerManager(workers);
		manager.start();

		thread = new Thread(new Runnable() {

			@Override
			public void run() {
				schedule();
			}

		}, "TaskExecutorScheduler");
		thread.start();

	}

	private void schedule() {
		while (!quit) {
			lock.lock();
			try {

				while (getAvailableTaskWorker() == null || tasks.isEmpty()) {
					try {
						cond.await(5, TimeUnit.MILLISECONDS);
					} catch (InterruptedException e) {
					}

					if (quit)
						return;
				}

			} finally {
				lock.unlock();
			}

			while (true) {
				TaskWorker worker = getAvailableTaskWorker();
				if (worker != null && !tasks.isEmpty()) {
					TaskWrapper wrapper = null;
					taskLock.lock();
					try {
						wrapper = tasks.poll();
					} finally {
						taskLock.unlock();
					}
					worker.run(wrapper);
				} else {
					break;
				}
			}
		}
	}

	private TaskWorker getAvailableTaskWorker() {
		for (TaskWorker worker : workers) {
			if (worker.getStatus() == TaskWorker.Status.IDEL)
				return worker;
		}
		return null;
	}

	@Override
	public void stop() {

		quit = true;
		lock.lock();
		try {
			cond.signal();
		} finally {
			lock.unlock();
		}

		manager.stop();

		for (int i = 0; i < workers.length; i++) {
			workers[i].stop();
		}

		if (thread != null) {
			try {
				thread.join();
			} catch (InterruptedException e) {
			}
		}

	}

}
